package com.lh.test;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;



import java.util.Properties;

public class Zksql {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");
        DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("OBS_BASE_LOG", new SimpleStringSchema(), properties));
        SingleOutputStreamOperator<JSONObject> page = stream.map(x -> {
            return JSON.parseObject(x);
        }).filter(x -> {
            return !StringUtils.isEmpty(x.getString("page"));
        });
        SingleOutputStreamOperator<Tuple3<String, String, String>> map = page.map(new MapFunction<JSONObject, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(JSONObject jsonObject) throws Exception {
                return Tuple3.of(jsonObject.getJSONObject("common").getString("os"),
                        jsonObject.getJSONObject("common").getString("mid"),
                        jsonObject.getJSONObject("page").getString("page_id")
                );
            }
        });
        map.print();
        env.execute();
    }
}
